package com.somg.kafka_study.demo_01;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerDemo extends Thread {
	
	private  final ConsumerConnector consumer;
	private final  String topic;
	private ExecutorService  executor;
	
	
	
	public ConsumerDemo(String topic)
	{
		consumer  = Consumer.createJavaConsumerConnector(createConsumerConfig());
		
		this.topic = topic;
		
	}
	
	public void shutdown()
	{
		if(consumer !=null)
		{
			consumer.shutdown();
		}
		if(executor != null)
		{
			executor.shutdown();
		}
		
	}
	
	private ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect","localhost:2181");
		props.put("group.id", "1");
		props.put("zookeeper.session.timeout.ms", "400");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");

		return new ConsumerConfig(props);
	}

	
	
	
	@Override
	public void run() {
		
		Map<String,Integer> topicmap = new HashMap<String, Integer>();
		topicmap.put(topic, 1);
		
		Map<String, List<KafkaStream<byte[], byte[]>>> streamMap = consumer.createMessageStreams(topicmap);
		
		KafkaStream<byte[] , byte[]>  stream  = streamMap.get(topic).get(0);
		
		ConsumerIterator<byte[],byte[]> it = stream.iterator();
		
		while(it.hasNext())
		{
			
			System.out.println("get data "+ new String(it.next().message()) );
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
		}
		
	}

	public static void main(String[] args) {
		
		ConsumerDemo demo  = new ConsumerDemo("test");
		demo.start();
		
		
		
	}
	
	
}
